package com.example.kafka.service;

import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * 演示整批次最后位点提交的错误示例
 */
@Service
@RequiredArgsConstructor
public class BatchOffsetCommitService {
    private final Logger log = LoggerFactory.getLogger(getClass());

    private final KafkaConsumer<String, String> consumer;

    /**
     * 用例1: 批量订单处理 - 整批次最后位点提交
     * 错误示例：不管中间处理是否成功，都提交整批次的最后位点
     */
    public String processBatchOrders(String ordersData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            Map<String, Long> lastSuccessOffsets = new HashMap<>();

            for (ConsumerRecord<String, String> record : records) {
                try {
                    processOrderLogic(record.value());
                    // 记录每个分区最后成功的位点
                    lastSuccessOffsets.put(record.topic() + "-" + record.partition(), record.offset());
                } catch (Exception e) {
                    log.error("Error processing order: {}", record.value(), e);
                    // 错误：继续处理下一条记录，没有正确处理失败的情况
                }
            }

            // 错误：直接提交整批次的最后位点
            consumer.commitSync();
            return "Batch orders processed";
        } catch (Exception e) {
            log.error("Error processing batch orders", e);
            throw new RuntimeException("Failed to process batch orders", e);
        }
    }

    /**
     * 用例2: 批量支付处理 - 整批次最后位点提交
     * 错误示例：不管中间处理是否成功，都提交整批次的最后位点
     */
    public String processBatchPayments(String paymentsData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    processPaymentLogic(record.value());
                } catch (Exception e) {
                    log.error("Error processing payment: {}", record.value(), e);
                    // 错误：继续处理下一条记录，没有正确处理失败的情况
                }
            }

            // 错误：直接提交整批次的最后位点
            consumer.commitSync();
            return "Batch payments processed";
        } catch (Exception e) {
            log.error("Error processing batch payments", e);
            throw new RuntimeException("Failed to process batch payments", e);
        }
    }

    /**
     * 用例3: 批量库存更新 - 整批次最后位点提交
     * 错误示例：不管中间处理是否成功，都提交整批次的最后位点
     */
    public String updateBatchInventory(String inventoryData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    updateInventoryLogic(record.value());
                } catch (Exception e) {
                    log.error("Error updating inventory: {}", record.value(), e);
                    // 错误：继续处理下一条记录，没有正确处理失败的情况
                }
            }

            // 错误：直接提交整批次的最后位点
            consumer.commitSync();
            return "Batch inventory updated";
        } catch (Exception e) {
            log.error("Error updating batch inventory", e);
            throw new RuntimeException("Failed to update batch inventory", e);
        }
    }

    /**
     * 用例4: 批量用户注册 - 整批次最后位点提交
     * 错误示例：不管中间处理是否成功，都提交整批次的最后位点
     */
    public String processBatchUserRegistrations(String usersData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    processUserRegistrationLogic(record.value());
                } catch (Exception e) {
                    log.error("Error processing user registration: {}", record.value(), e);
                    // 错误：继续处理下一条记录，没有正确处理失败的情况
                }
            }

            // 错误：直接提交整批次的最后位点
            consumer.commitSync();
            return "Batch user registrations processed";
        } catch (Exception e) {
            log.error("Error processing batch user registrations", e);
            throw new RuntimeException("Failed to process batch user registrations", e);
        }
    }

    /**
     * 用例5: 批量日志处理 - 整批次最后位点提交
     * 错误示例：不管中间处理是否成功，都提交整批次的最后位点
     */
    public String processBatchLogs(String logsData) {
        try {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                try {
                    processLogLogic(record.value());
                } catch (Exception e) {
                    log.error("Error processing log: {}", record.value(), e);
                    // 错误：继续处理下一条记录，没有正确处理失败的情况
                }
            }

            // 错误：直接提交整批次的最后位点
            consumer.commitSync();
            return "Batch logs processed";
        } catch (Exception e) {
            log.error("Error processing batch logs", e);
            throw new RuntimeException("Failed to process batch logs", e);
        }
    }

    // 模拟业务处理逻辑
    private void processOrderLogic(String orderData) {
        // 实际订单处理逻辑
    }

    private void processPaymentLogic(String paymentData) {
        // 实际支付处理逻辑
    }

    private void updateInventoryLogic(String inventoryData) {
        // 实际库存更新逻辑
    }

    private void processUserRegistrationLogic(String userData) {
        // 实际用户注册处理逻辑
    }

    private void processLogLogic(String logData) {
        // 实际日志处理逻辑
    }
}
